package com.yesep.upstream;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import com.alibaba.dubbo.common.utils.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @program: gateway
 * @description: 上行线程
 * @author:quliang
 * @create: 2019-07-17 11:06
 **/
public class UpStreamDataThread extends Thread {

    private static final Logger logger = LoggerFactory.getLogger(UpStreamDataThread.class);

    /**
     * 上报消息队列
     */
    private static BlockingQueue<Object> upQueue = new LinkedBlockingQueue<>();
    /**
     * 处理的消息数量
     */
    private long msgCount = 0;

    /**
     * 是否已启动
     */
    private boolean isStarted = false;

    // 上报数据处理任务 线程池
    final ExecutorService cachedThreadPool = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors() * 2,
            new NamedThreadFactory("UpStreamThreadPool"));

    public UpStreamDataThread(String threadName) {
        this.setName(threadName);

    }

    public static void put(Object data) {
        upQueue.add(data);
    }

    public static void put(List<Object> dataList) {
        for (Object data : dataList) {
            upQueue.add(data);
        }
    }

    public void init() {
        try {
            if (!isStarted) {
                this.start();
                isStarted = true;
            }
        } catch (Exception e) {
            logger.error("UpStreamDataThread Error", e);
        }
    }

    @Override
    public void run() {
        Object usd;
        try {
            for (; ; ) {
                // 从上行的队列中获取上行数据，然后创建上行任务，放到线程池去执行
                if ((usd = upQueue.take()) != null) {
                    cachedThreadPool.execute(new UpStreamDataTask(usd));
                    msgCount++;
                }
                Thread.sleep(0);
            }
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * @param data
     * @Description: 把数据添加到上行队列中
     * @date 2017年7月6日, 上午12:25:37
     * @author chenyun
     * @version 1.0.0
     */
    public static void putUpStreamDataFromServerHandler(Object data) {
        if (data == null)
            return;
        put(data);
    }
}
